package com.xiaojie.init;

import com.rabbitmq.client.AMQP;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.entity.ContentType;
import org.apache.http.impl.client.HttpClients;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.BeanUtils;
import org.springframework.util.StringUtils;

import javax.xml.bind.DatatypeConverter;
import java.io.IOException;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

@Slf4j
public abstract class RabbitHandlerUtils {

    private final static String DELAYED_TYPE = "x-delayed-type";

    private final static String DELAYED_MESSAGE = "x-delayed-message";

    /**
     * 创建交换机
     *
     * @param rabbitExchange 创建交换机的对象
     * @return 交换机
     */
    public static Exchange createExchange(RabbitExchange rabbitExchange) {
        RabbitAdmin admin = RabbitContextHolder.getRabbitAdmin(rabbitExchange.getVhost());
        Exchange exchange = initExchange(rabbitExchange);
        admin.declareExchange(exchange);
        return exchange;
    }

    /**
     * 移除交换机
     *
     * @param vhostName    虚拟主机名
     * @param exchangeName 交换机的名称
     * @return
     */
    public static boolean deleteExchange(String vhostName, String exchangeName) {
        RabbitAdmin rabbitAdmin = RabbitContextHolder.getRabbitAdmin(vhostName);
        return rabbitAdmin.deleteExchange(exchangeName);
    }


    /**
     * 创建队列
     *
     * @param rabbitQueue
     * @return
     */
    public static Queue createQueue(RabbitQueue rabbitQueue) {
        RabbitAdmin rabbitAdmin = RabbitContextHolder.getRabbitAdmin(rabbitQueue.getVhost());
        String queueName = rabbitQueue.getName();
        Queue queue = new Queue(queueName);

        if (queueExist(queueName, rabbitAdmin)) {
            throw new RuntimeException("The queue " + rabbitQueue.toString() + " 已经存在。");
        }
        BeanUtils.copyProperties(rabbitQueue, queue);
        rabbitAdmin.declareQueue(queue);
        return queue;
    }

    /**
     * 移除队列
     *
     * @param vhostName 虚拟主机名
     * @param queueName 队列名
     */
    public static void deleteQueue(String vhostName, String queueName) {
        RabbitAdmin rabbitAdmin = RabbitContextHolder.getRabbitAdmin(vhostName);
        rabbitAdmin.deleteQueue(queueName);
    }


    /**
     * 创建绑定关系
     */
    public static void bind(RabbitBinding binding) {
        RabbitAdmin rabbitAdmin = RabbitContextHolder.getRabbitAdmin(binding.getVhost());
        Binding b = new Binding(binding.getQueue(),
                Binding.DestinationType.QUEUE,
                binding.getExchange(),
                binding.getRoutingKey(),
                binding.getArguments());
        rabbitAdmin.declareBinding(b);
    }

    /**
     * 解绑操作
     */
    public static void unbind(RabbitBinding binding) {
        RabbitAdmin rabbitAdmin = RabbitContextHolder.getRabbitAdmin(binding.getVhost());
        Binding b = new Binding(binding.getQueue(),
                Binding.DestinationType.QUEUE,
                binding.getExchange(),
                binding.getRoutingKey(),
                binding.getArguments());
        rabbitAdmin.removeBinding(b);
    }

    /**
     * 清空队列
     *
     * @param vhostName 虚拟主机名
     * @param queueName 队列名
     * @param noWait    是否等待，true是异步清空，false是同步清空
     */
    public static void purgeQueue(String vhostName, String queueName, boolean noWait) {
        RabbitAdmin rabbitAdmin = RabbitContextHolder.getRabbitAdmin(vhostName);
        rabbitAdmin.purgeQueue(queueName, noWait);
    }


    /**
     * 获取到消息的数量
     *
     * @param vhostName 虚拟主机名
     * @param queueName 队列名
     * @return 队列中消息的数量
     */
    public static int getMessageCount(String vhostName, String queueName) {
        RabbitAdmin rabbitAdmin = RabbitContextHolder.getRabbitAdmin(vhostName);
        if (StringUtils.isEmpty(queueName)) {
            throw new RuntimeException("Queue name can not be null");
        }
        Integer messageCount = rabbitAdmin.getRabbitTemplate().execute(channel -> {
            try {
                AMQP.Queue.DeclareOk declareOk = channel.queueDeclarePassive(queueName);
                return declareOk.getMessageCount();
            } catch (Exception e) {
                log.error("获取队列消息识别[{}]", queueName, e);
                return -1;
            }
        });
        return messageCount == null ? 0 : messageCount;
    }

    /**
     * 判断队列是否存在
     *
     * @param queueName   队列名
     * @param rabbitAdmin 某连接配置下的{@link RabbitAdmin}
     * @return true表示存在，false表示不存在
     */
    private static boolean queueExist(String queueName, RabbitAdmin rabbitAdmin) {
        String name = rabbitAdmin.getRabbitTemplate().execute(channel -> {
            try {
                //若找不到，直接会抛出404的错误
                AMQP.Queue.DeclareOk declareOk = channel.queueDeclarePassive(queueName);
                return declareOk.getQueue();
            } catch (Exception e) {
                log.error("查询异常", e);
                return null;
            }
        });
        return name != null && !"".equals(name);
    }

    /**
     * 调用Mq创建虚拟主机
     *
     * @param vhost 虚拟主机配置
     * @throws IOException
     */
    public static void createVHost(RabbitVirtualHost vhost) throws IOException {
        HttpClient client = HttpClients.createDefault();
        HttpPut httpPut = new HttpPut();
        String authTemplate = "%s:%s";
        String authString = String.format(authTemplate, vhost.getUsername(), vhost.getPassword());
        String encoding = DatatypeConverter.printBase64Binary(authString.getBytes(StandardCharsets.UTF_8));
        httpPut.setHeader("content-type", ContentType.APPLICATION_JSON.toString());
        httpPut.setHeader("Authorization", "Basic " + encoding);
        String hostTemplate = "http://%s:%d";
        String apiTemplate = "http://%s:%d/api/vhosts/%s";
        String host = String.format(hostTemplate, vhost.getHost(), 15672);
        String api = String.format(apiTemplate, vhost.getHost(), 15672, vhost.getVhost());

        httpPut.setURI(URI.create(api));
        HttpResponse response = client.execute(HttpHost.create(host), httpPut);
        log.info("创建Rabbit虚拟主机的配置 : " + api + " : " + response);

//        Assert.assertTrue(response.getStatusLine().getStatusCode() == 200 ||
//                response.getStatusLine().getStatusCode() == 201
//                || response.getStatusLine().getStatusCode() == 204);
    }

    /**
     * 初始化交换机
     */
    private static Exchange initExchange(RabbitExchange rabbitExchange) {
        //判断是否是延迟队列
        if (rabbitExchange.isDelayed()) {
            //定义延迟队列
            Map<String, Object> arguments = new HashMap<>();
            arguments.put(DELAYED_TYPE, rabbitExchange.getType().name().toLowerCase());
            return new CustomExchange(rabbitExchange.getName(),
                    DELAYED_MESSAGE,
                    rabbitExchange.isDurable(),
                    rabbitExchange.isAutoDelete(),
                    arguments);
        }
        switch (rabbitExchange.getType()) {
            case DIRECT:  //直连模式
                return new DirectExchange(rabbitExchange.getName(),
                        //交换机是否持久化
                        rabbitExchange.isDurable(),
                        //当所有的绑定关系被删除时，自动删除队列
                        rabbitExchange.isAutoDelete(),
                        //交换器的其他参数，可以为空
                        rabbitExchange.getArguments());
            case TOPIC:  //通配符模式
                return new TopicExchange(rabbitExchange.getName(),
                        rabbitExchange.isDurable(),
                        rabbitExchange.isAutoDelete(),
                        rabbitExchange.getArguments());
            case FANOUT:  //广播模式
                return new FanoutExchange(rabbitExchange.getName(),
                        rabbitExchange.isDurable(),
                        rabbitExchange.isAutoDelete(),
                        rabbitExchange.getArguments());
            case HEADER: //该类型不常见
                return new HeadersExchange(rabbitExchange.getName(),
                        rabbitExchange.isDurable()
                        , rabbitExchange.isAutoDelete(),
                        rabbitExchange.getArguments());
            default:
                return null;
        }
    }
}
